import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.elasticsearch.spark._

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/10/14 09:40
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object modify_price_taobao {

  def main(args: Array[String]): Unit = {

  val conf = new SparkConf()
  conf.setAppName(s"${this.getClass.getSimpleName}")
  conf.set("spark.debug.maxToStringFields", "500")
  conf.setMaster("local[*]")
  conf.set("es.nodes", "192.168.1.157")
  conf.set("es.port", "9200")
  conf.set("cluster.name", "O2OElastic")
  conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  conf.set("spark.sql.caseSensitive", "true")
  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)
  sc.setLogLevel("WARN")
  sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
  sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
  sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

  val values: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
    """
      |{
      |  "query": {
      |    "term": {
      |      "good_id": {
      |        "value": "609069166286"
      |      }
      |    }
      |  }
      |}
    """.stripMargin).values

  val frame: DataFrame = sqlContext.read.json(values)
    frame.selectExpr("priceText","salesAmount").show(false)


    frame.createOrReplaceTempView("frame")
  //frame.write.json("s3a://private/private_han/TM/2019/9/modify_taobao")
    val frame1 =  sqlContext.sql(
          """
            |select
            |*,
            |(sellCount*price1) as sales
            |from
            |(select
            |*,
            |case
            |when good_id ='609069166286' then 20
            |else priceText end as price1
            |from
            |frame)
          """.stripMargin).drop("priceText","salesAmount")
            .withColumnRenamed("price1","priceText")
            .withColumnRenamed("sales","salesAmount")

    frame1.selectExpr("priceText","salesAmount").show(false)

  val result = frame1.toJSON.rdd.map(line => {
    val lines = JSON.parseObject(line)
    val evaluates = lines.getOrDefault("evaluates", "-1").toString
    val Base_Info = lines.getOrDefault("Base_Info", "-1").toString
    var ev = "-1"
    var str = new JSONObject
    if (evaluates.equals("-1")) {
      ev = "{\"fuyi\":\"-1\"}"
    } else {
      str = JSON.parseObject(evaluates)
    }
    if (!ev.contains("fuyi")) {
      lines.put("evaluates", str)
    } else {
      val evs = JSON.parseObject(ev)
      lines.put("evaluates", evs)
    }
    lines.put("Base_Info",Base_Info)
    lines
  })
  result.saveToEs(s"2020_tmall/tmall_2020_3", Map("es.mapping.id" -> "good_id"))
}

}
